package com.atguigu.edu.app.dws;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.bean.TradeSkuOrderBean;
import com.atguigu.edu.util.DateFormatUtil;
import com.atguigu.edu.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;


import java.time.Duration;

public class DwsInteractionOrderWindow {

    public static void main(String[] args) throws Exception {
        //1.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //  设置并行度
        env.setParallelism(4);
        //2.检查点相关设置  略
        //3.从kafka的dwd_interaction_comment主题中读取数据
        //3.1声明消费的主题以及消费者组
        String topic = "dwd_interaction_comment";
        String groupId = "dws_interaction_comment_group";
        //3.2创建消费者对象
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        //3.3消费数据   封装成流
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);

        //4.对读取的数据进行类型转换   jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);
        jsonObjDS.print("&&&&");
        //5.对流中的数据进行类型转换   jsonObj  ->  实体类
        SingleOutputStreamOperator<TradeSkuOrderBean> orderBeanDS = jsonObjDS.map(new MapFunction<JSONObject, TradeSkuOrderBean>() {
            @Override
            public TradeSkuOrderBean map(JSONObject jsonObj) throws Exception {
                String courseId = jsonObj.getString("course_id");
                String userId = jsonObj.getString("user_id");
                Long reviewStars = jsonObj.getLong("review_stars");
                Long ts = jsonObj.getLong("ts") * 1000L;
                TradeSkuOrderBean bbb = TradeSkuOrderBean.builder()
                        .course_id(courseId)
                        .user_id(userId)
                        .review_stars(reviewStars)
                        .orderCount(1L)
                        .userCount(1L)
                        .pfCount(1L)
                        .gCount(1L)
                        .ts(ts)
                        .build();
                return bbb;

            }
        });

        //6.指定Watermark以及提取时间时间字段
        SingleOutputStreamOperator<TradeSkuOrderBean> withWatermarkDS = orderBeanDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TradeSkuOrderBean>forMonotonousTimestamps()
                .withTimestampAssigner(
                        new SerializableTimestampAssigner<TradeSkuOrderBean>() {
                            @Override
                            public long extractTimestamp(TradeSkuOrderBean element, long recordTimestamp) {
                                return element.getTs();
                            }
                        }
                )
        );
        //7.按照课程维度进行分组
        KeyedStream<TradeSkuOrderBean, String> courseIdDS = withWatermarkDS.keyBy(TradeSkuOrderBean::getCourse_id);
        courseIdDS.print("****");

        WindowedStream<TradeSkuOrderBean, String, TimeWindow> windowDS = courseIdDS.window(TumblingEventTimeWindows.of(Time.seconds(1L)));

        //8.聚合操作
        SingleOutputStreamOperator<TradeSkuOrderBean> reduceDS = windowDS.reduce(
                new ReduceFunction<TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean reduce(TradeSkuOrderBean value1, TradeSkuOrderBean value2) throws Exception {
                        value1.setOrderCount(value1.getOrderCount() + value2.getOrderCount());
                        value1.setUserCount(value1.getUserCount() + value2.getUserCount());
                        value1.setPfCount(value1.getPfCount() + value2.getPfCount());
                        value1.setGCount(value1.getGCount() + value2.getGCount());
                        return value1;
                    }
                },
//                new ProcessWindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
////                    @Override
////                    public void process(String s, ProcessWindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>.Context context, Iterable<TradeSkuOrderBean> elements, Collector<TradeSkuOrderBean> out) throws Exception {
////                        for (TradeSkuOrderBean orderBean : elements) {
////                            orderBean.setStt(DateFormatUtil.toYmdHms(context.window().getStart()));
////                            orderBean.setEdt(DateFormatUtil.toYmdHms(context.window().getEnd()));
////                            orderBean.setTs(System.currentTimeMillis());
////                            out.collect(orderBean);
////                        }
////                    }
////                }
                new ProcessWindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<TradeSkuOrderBean> elements, Collector<TradeSkuOrderBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(context.window().getStart());
                        String edt = DateFormatUtil.toYmdHms(context.window().getEnd());
                        for (TradeSkuOrderBean orderBean : elements) {
                            orderBean.setStt(stt);
                            orderBean.setEdt(edt);
                            orderBean.setTs(System.currentTimeMillis());
                            System.out.println(orderBean);
                            out.collect(orderBean);
                        }
                    }
                }

        );
        //测试
        reduceDS.print(">>>>>");
        //将异步I/O操作应用于DataStream作为DataStream的一次转换操作
/*        AsyncDataStream.unorderedWait(
                reduceDS,
                //执行的异步操作   实现分发请求的AsyncFunction
                new DimAsyncFunction<TradeSkuOrderBean>("课程表名")
        )*/
        env.execute();
    }
}
